package neiltwist.cfd.akka;

import static akka.actor.Actors.actorOf;
import static akka.actor.Actors.poisonPill;
import akka.actor.ActorRef;
import akka.actor.UntypedActor;
import akka.actor.UntypedActorFactory;
import akka.actor.UntypedChannel;
import akka.routing.Routing.Broadcast;

public class Master extends UntypedActor {

	private ActorRef router;
	private UntypedChannel channel;

	private float[][] faw;

	public Master(final int nrOfWorkers) {

		// create the workers
		final ActorRef[] workers = new ActorRef[nrOfWorkers];
		for (int i = 0; i < nrOfWorkers; i++) {
			workers[i] = actorOf(Worker.class).start();
		}

		// wrap them with a load-balancing router
		router = actorOf(new UntypedActorFactory() {
			public UntypedActor create() {
				return new Router(workers);
			}
		}).start();
	}

	// message handler
	public void onReceive(Object message) {
		if (message instanceof CalculateDiffuse) {
			//cache sender...
			channel = getContext().getChannel();

			CalculateDiffuse cd = (CalculateDiffuse)message;

			float a = cd.getA();
			
			int N = cd.getN();
			faw = new float[N+2][N+2];

			// schedule work
			for ( int i=1 ; i<=N ; i++ ) {
				for ( int j=1 ; j<=N ; j++ ) {
					// We need: 
					// a,
					float x0 = cd.getX0().get(i, j);
					float x1 = cd.getX().get(i-1, j);
					float x2 = cd.getX().get(i+1, j);
					float x3 = cd.getX().get(i, j-1); // x3 = x i   j-1,
					float x4 = cd.getX().get(i, j+1); // x4 = x i   j+1;
					router.tell(new Work(a,x0,x1,x2,x3,x4,i,j), getContext());
				}
			}
			
			router.tell(new Finished(),getContext());

		} else if (message instanceof Result) {
			// re-save individual result
			Result result = (Result)message;
			faw[result.getI()] [result.getJ()] = result.getValue();
		} else if (message instanceof Finished) {
			// finished, so respond
			channel.tryTell(faw);
		} else throw new IllegalArgumentException("Unknown message [" + message + "]");
	}

	@Override
	public void preStart() {
		// No Op
	}

	@Override
	public void postStop() {
		// send a PoisonPill to all workers telling them to shut down themselves
		router.tell(new Broadcast(poisonPill()));

		// send a PoisonPill to the router, telling him to shut himself down
		router.tell(poisonPill());
	}

}
